package com.github.ghsea.scheduler.core;

import com.github.ghsea.scheduler.dto.Task;
import com.github.ghsea.scheduler.dto.TaskExecutionEvent;
import com.github.ghsea.scheduler.entity.AliveTaskStatistics;
import com.github.ghsea.scheduler.entity.TaskEntity;
import com.github.ghsea.scheduler.service.TaskSchedulerService;
import com.github.ghsea.scheduler.service.TaskService;
import com.github.ghsea.scheduler.timer.HashedWheelTimer;
import com.github.ghsea.scheduler.timer.Timeout;
import com.github.ghsea.scheduler.timer.Timer;
import com.github.ghsea.scheduler.timer.TimerTask;
import com.github.ghsea.scheduler.util.NetUtils;
import com.github.ghsea.scheduler.util.RunnableDelegate;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import java.time.Duration;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import javax.annotation.Resource;
import lombok.Builder;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;


/**
 * @author guhai
 */
@Component
@Slf4j
public class TaskScheduler implements ApplicationListener<ApplicationReadyEvent> {

    /**
     * ticks需要是2的N次方
     */
    private static int TICKS = 2048;

    /**
     * 每个timer最多执行的task数目，超过该数目，将启动新的timer
     */
    private static final int TASK_PER_TIMER = 10000;

    private int timerCount = 0;

    /**
     * task编号
     */
    private AtomicInteger taskNum = new AtomicInteger(0);

    private final Object lock = new Object();

    /**
     * 为减少数据堆积到个别tick的情况，task进入时，先放入队列，然后定时从队列拉取将其放入Timer
     */
    private ConcurrentLinkedQueue<Task> pendingTasks = new ConcurrentLinkedQueue<>();

    @Resource(name = "tinyTaskExecutor")
    private ExecutorService tinyTaskExecutor;

    @Resource(name = "scheduledExecutor")
    private ScheduledExecutorService scheduledExecutor;

    @Resource
    private TaskExecutor taskExecutor;

    protected final String serverName = NetUtils.getLocalHost();

    @Autowired
    private TaskService taskService;

    final int dbFetchSize = 1000;

    private ConcurrentHashMap<String, Timeout> bizCode2Timeout = new ConcurrentHashMap<>();

    private CacheLoader<String, Long> cacheLoader = new TaskExpireTimeCacheLoader();
    private Cache<String, Long> expireTimeCache = CacheBuilder.newBuilder().refreshAfterWrite(Duration.ofMinutes(1L)).build(cacheLoader);

    private ConcurrentHashMap<RoundConfiguration, CopyOnWriteArrayList<TimerWrapper>> duration2Wrappers = new ConcurrentHashMap<>();

    /**
     * 维持心跳的时间间隔
     */
    final int HEARTBEAT_INTERVAL_SECONDS = 30;

    /**
     * 判断机器是否宕机的心跳时间间隔
     */
    private final int SERVER_CRASH_JUDGE_SECONDS = 30 * 10;

    @Autowired
    private TaskSchedulerService taskSchedulerService;

    private ListeningExecutorService cacheLoaderService;

    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {
        cacheLoaderService = MoreExecutors.listeningDecorator(tinyTaskExecutor);
        initCache();
        initTaskAndSchedule(serverName, false);
        pollTaskByTickDuration();

        doHeartbeat();

        log.info("init Scheduler and prepared to work");
    }

    /**
     * 从数据库加载Task,任加入Timer调度
     *
     * @param serverId 用于标识相关联的task
     * @param isDiscovery 是否恢复模式。true:恢复模式，false:初始化模式
     */
    protected void initTaskAndSchedule(String serverId, boolean isDiscovery) {
        String mode = isDiscovery ? "discovery" : "init";
        log.info(mode + " tasks from database start,serverName = {}", serverId);

        AliveTaskStatistics statistics = taskService.aliveTaskStatistics(serverId);
        int aliveTasksCount = statistics.getAliveTasksCount();
        if (aliveTasksCount == 0) {
            if (!isDiscovery) {
                log.error("Can not find task for serverName={}.", serverId);
            }
            /**
             * TODO isDiscovery模式下将task_scheduler中的相关server的记录置为不可用，
             * 以防该server的任务全部被接管后仍然被认为机器已宕机,再次触发discovery操作
             */
            return;
        }

        int taskCountPerTick = (int) Math.ceil((double) aliveTasksCount / TICKS);
        tinyTaskExecutor.execute(() -> {
            long startId = 0;
            int totalTaskCount = 0;
            List<TaskEntity> tasks;
            long startMs = System.currentTimeMillis();
            while (true) {
                tasks = taskService.fetchAliveTasks(serverId, startId, dbFetchSize);
                totalTaskCount = totalTaskCount + tasks.size();
                List<List<TaskEntity>> taskPages = Lists.partition(tasks, taskCountPerTick);
                for (List<TaskEntity> taskPage : taskPages) {
                    for (TaskEntity task : taskPage) {
                        startId = Math.max(startId, task.getId());
                        schedule(task.toTask());
                    }
                }

                if (!serverName.equals(serverId)) {
                    tasks.forEach(e -> e.setServer(serverName));
                    taskService.batchUpdateServerByTaskId(tasks);
                }
                log.info("loaded {} tasks ,startId={}", tasks.size(), startId);

                try {
                    //降低对数据库的压力
                    TimeUnit.MILLISECONDS.sleep(10);
                } catch (InterruptedException e) {
                    log.error(e.getMessage(), e);
                }

                if (tasks.size() < dbFetchSize) {
                    if (!isDiscovery) {
                        /**
                         * TODO isDiscovery模式下将task_scheduler中的相关server的记录置为不可用，
                         * 以防该server的任务全部被接管后仍然被认为机器已宕机,再次触发discovery操作
                         */
                    }

                    break;
                }
            }

            long endMs = System.currentTimeMillis();

            finishLoadTask(serverId);
            //TODO totalTaskCount要重新算
            log.info(mode + " {} tasks for {} from database.elapsed time {} ms", totalTaskCount, serverId, endMs - startMs);
        });
    }

    private TimerWrapper createTimerIfBusy(int intervalSeconds) {
        RoundConfiguration roundConfiguration = RoundConfiguration.of(intervalSeconds);
        duration2Wrappers.putIfAbsent(roundConfiguration, new CopyOnWriteArrayList<>());
        //如果不存在roundConfiguration所对应的Timer，则新建一个Timer
        CopyOnWriteArrayList<TimerWrapper> timers = duration2Wrappers.get(roundConfiguration);
        if (timers.size() == 0) {
            synchronized (lock) {
                if (timers.size() == 0) {
                    timers.add(newTimer(timerCount++, roundConfiguration));
                }
            }
        }

        //如果Timer都太忙，则再新建一个
        TimerWrapper timerWrapper = timers.get(timers.size() - 1);
        if (timerWrapper.isBusy()) {
            synchronized (lock) {
                if (timers.get(timers.size() - 1).isBusy()) {
                    timers.add(newTimer(timerCount++, roundConfiguration));
                    timerWrapper = timers.get(timers.size() - 1);
                }
            }
        }
        return timerWrapper;
    }

    private void pollTaskByTickDuration() {
        scheduledExecutor.scheduleAtFixedRate(new RunnableDelegate(() -> {
                    Task task = pendingTasks.poll();
                    if (task == null) {
                        return;
                    }

                    TimerWrapper timerWrapper = createTimerIfBusy(task.getIntervalSeconds());
                    Timer timer = timerWrapper.getTimer();
                    //第一次执行时，不需要等待入参的监控频率，因为在监控频率比较长的情况下，第一个dealy也会很长，看起来就像是丢失一了一个点一样.
                    long firstDelay = timerWrapper.getRoundConfig().tickDurationNano();
                    log.info("poll and begin scheduling: [{}] ", task.toString());
                    Timeout timeout = timer.newTimeout(new ExtTimerTask(timerWrapper, taskNum.getAndIncrement(), task), firstDelay, TimeUnit.NANOSECONDS);
                    bizCode2Timeout.put(task.getBizCode(), timeout);
                    timerWrapper.increaseTaskCount();
                }),
                TimeUnit.MINUTES.toMillis(1),
                TimeUnit.NANOSECONDS.toMillis(RoundConfiguration.ONE_MINUTE.tickDurationNano()),
                TimeUnit.MILLISECONDS);
    }

    public void schedule(Task task) {
        //Task的执行时间小于环的一圈的时间间隔，会导致Task局部聚集
        int minRoundDuration = RoundConfiguration.HALF_MINUTE.roundDurationSec;
        if (task.getIntervalSeconds() < minRoundDuration) {
            log.warn("reset intervalSeconds for {}", task);
            task.setIntervalSeconds(minRoundDuration);
        }

        log.info("pending and wait scheduling:[{}] ", task.toString());
        pendingTasks.offer(task);

        expireTimeCache.put(task.getBizCode(), task.getStopTime().getTime());
    }

    /**
     * TODO
     *
     * @param taskId task.getBizCode()
     */
    public void stop(String taskId, Date stopTime) {
        expireTimeCache.put(taskId, stopTime.getTime());
        Timeout timeout = bizCode2Timeout.get(taskId);
        if (timeout != null) {
            log.info("stopped the task:{}" + taskId);
            timeout.cancel();
        } else {
            log.info("did not to stop the task:{}" + taskId);
        }
    }

    private TimerWrapper newTimer(int number, RoundConfiguration roundConfig) {
        log.info("create Timer{}", number);
        HashedWheelTimer timer = new HashedWheelTimer(roundConfig.tickDurationNano(), TimeUnit.NANOSECONDS, roundConfig.ticks);
        return TimerWrapper.builder()
                .roundConfig(roundConfig)
                .timer(timer)
                .timerNum(number)
                .taskCount(new AtomicInteger(0))
                .build();
    }

    @Data
    @Builder
    private static class TimerWrapper {

        private final Timer timer;

        /**
         * timer编号
         */
        private int timerNum;

        /**
         * 该Timer持有的Task数量
         */
        private AtomicInteger taskCount;

        private RoundConfiguration roundConfig;

        private void increaseTaskCount() {
            taskCount.incrementAndGet();
        }

        private boolean isBusy() {
            return taskCount.intValue() >= TaskScheduler.TASK_PER_TIMER;
        }
    }

    private class ExtTimerTask implements TimerTask {

        private TimerWrapper timerWrapper;

        int taskCount;

        private int loop = 0;

        private Task queryTask;

        /**
         * 上一次执行的时间
         */
        private long lastExeTimeNano;

        private Timeout firstTimeout;

        private final Object lock = new Object();

        ExtTimerTask(TimerWrapper timerWrapper, int taskCount, Task queryTask) {
            this.timerWrapper = timerWrapper;
            this.taskCount = taskCount;
            this.queryTask = queryTask;
        }

        @Override
        public Timeout getFirstTimeout() {
            return firstTimeout;
        }

        private void setFirstTimeout(Timeout timeout) {
            if (firstTimeout == null) {
                synchronized (lock) {
                    if (firstTimeout == null) {
                        this.firstTimeout = timeout;
                    }
                }
            }
        }

        @Override
        public void run(Timeout timeout) {
            if (lastExeTimeNano == 0) {
                lastExeTimeNano = System.nanoTime();
            }

            if (isTaskExpired(queryTask.getBizCode())) {
                timerWrapper.getTaskCount().decrementAndGet();
                log.info("stop scheduler because of it's end time: [{}] ", queryTask.toString());
                return;
            }

            log.debug("debug running timerWrapper:{}.it's going to send msg", queryTask.toString());
            loop++;
            TaskExecutionEvent taskExecutionEvent = new TaskExecutionEvent()
                    .setLoop(loop)
                    .setTaskCount(taskCount)
                    .setTimerNum(timerWrapper.getTimerNum())
                    .setBizCode(queryTask.getBizCode())
                    .setExeTime(System.currentTimeMillis())
                    .setInterval(queryTask.getIntervalSeconds());
            taskExecutor.produceTaskEvent(taskExecutionEvent);

            long delay = TimeUnit.SECONDS.toNanos(queryTask.getIntervalSeconds());
            Timeout lastTimeout = timerWrapper.getTimer().newTimeout(this, delay, TimeUnit.NANOSECONDS);
            bizCode2Timeout.put(queryTask.getBizCode(), lastTimeout);
            setFirstTimeout(lastTimeout);

            lastExeTimeNano = System.nanoTime();
        }
    }

    enum RoundConfiguration {
        /**
         * 每圈半分钟,每个tick 58ms
         */
        HALF_MINUTE(30, 512),

        /**
         * 每圈1分钟,每个tick 58ms
         */
        ONE_MINUTE(60, 1024),

        /**
         * 每圈5分钟,每个tick 146ms
         */
        FIVE_MINUTE(300, 2048),

        /**
         * 每圈30分钟,每个tick(决定了精度) 439ms
         */
        HALF_HOUR(1800, 4096);

        /**
         * 每一圈的时间，单位秒
         */
        private int roundDurationSec;

        /**
         * 每一圈的步数
         */
        private int ticks;

        RoundConfiguration(int roundDurationSec, int ticks) {
            this.roundDurationSec = roundDurationSec;
            this.ticks = ticks;
        }

        static RoundConfiguration of(int durationSeconds) {
            long sixHour = TimeUnit.HOURS.toSeconds(6);
            if (durationSeconds <= ONE_MINUTE.roundDurationSec) {
                return HALF_MINUTE;
            } else if (durationSeconds <= HALF_HOUR.roundDurationSec) {
                return FIVE_MINUTE;
            } else if (durationSeconds <= sixHour) {
                //目前支持的最大监控频繁是6小时
                return HALF_HOUR;
            } else {
                throw new IllegalArgumentException("exceed the max duration " + sixHour + " seconds");
            }
        }

        /**
         * 每一个tick的时间间隔
         *
         * @return 每一个tick的时间间隔
         */
        long tickDurationNano() {
            return (long) Math.floor(TimeUnit.SECONDS.toNanos(roundDurationSec) / ticks);
        }
    }


    private class TaskExpireTimeCacheLoader extends CacheLoader<String, Long> {
        @Override
        public Long load(String key) {
            TaskEntity task = taskService.queryAliveTaskCode(key);
            return task != null ? task.getStopTime().getTime() : 0;
        }

        @Override
        public Map<String, Long> loadAll(Iterable<? extends String> keys) {
            log.info("start to load tasks into cache,server={}", serverName);

            long startId = 0;
            Map<String, Long> cache = new HashMap<>();

            while (true) {
                List<TaskEntity> aliveTasks = taskService.fetchAliveTasks(serverName, startId, dbFetchSize);
                if (aliveTasks == null || aliveTasks.size() == 0) {
                    break;
                }

                Map<String, Long> tasks = aliveTasks.stream().
                        collect(Collectors.toMap(TaskEntity::getBizCode, t -> t.getStopTime().getTime()));

                cache.putAll(tasks);

                if (aliveTasks.size() < dbFetchSize) {
                    break;
                }
            }

            log.info("loaded {} tasks into cache,server={}", cache.size(), serverName);
            return cache;
        }

        @Override
        public ListenableFuture<Long> reload(String key, Long oldValue) {
            return cacheLoaderService.submit(() -> load(key));
        }
    }


    /**
     * 判断task是否已过期或者已停上
     */
    private boolean isTaskExpired(String taskCode) {
        Long expireTime = expireTimeCache.getIfPresent(taskCode);
        return expireTime != null && expireTime < System.currentTimeMillis();
    }

    /**
     * 维持心跳
     */
    private void doHeartbeat() {
        scheduledExecutor.scheduleAtFixedRate(new RunnableDelegate(() -> {
            log.info("heartbeat by server:{}", serverName);
            long heartbeatTime = System.currentTimeMillis();
            taskSchedulerService.heartbeat(serverName, new Date(heartbeatTime),
                    new Date(nextHearBeatTime(heartbeatTime)));
        }), 5, HEARTBEAT_INTERVAL_SECONDS, TimeUnit.SECONDS);
    }

    private long nextHearBeatTime(long heartbeatTime) {
        return heartbeatTime + SERVER_CRASH_JUDGE_SECONDS;
    }

    private void initCache() {
        try {
            Map<String, Long> taskCache = cacheLoader.loadAll(null);
            expireTimeCache.putAll(taskCache);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    protected void finishLoadTask(String server){

    }

}
